package com.jwong.test.operator;

import io.fabric8.kubernetes.api.model.*;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.ExecListener;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.dsl.RollableScalableResource;
import io.fabric8.kubernetes.client.dsl.ServiceResource;
import io.fabric8.kubernetes.client.utils.Serialization;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.concurrent.*;

import static java.lang.String.format;

@ControllerConfiguration
public class MysqlMasterSlaveReconciler implements Reconciler<MySqlMasterSlave>, ErrorStatusHandler<MySqlMasterSlave> {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private final KubernetesClient kubernetesClient;

    private static final String MYSQL_CONNECTOR_COMMAND = "mysql-connector.sh";

    private static final String MASTER = "master";
    private static final String SLAVE = "slave";


    public MysqlMasterSlaveReconciler(KubernetesClient kubernetesClient) {
        this.kubernetesClient = kubernetesClient;
    }

    @Override
    public Optional<MySqlMasterSlave> updateErrorStatus(MySqlMasterSlave resource, RetryInfo retryInfo, RuntimeException e) {
        resource.getStatus().setErrorMessage("Error: " + e.getMessage());
        return Optional.of(resource);
    }

    @Override
    public UpdateControl<MySqlMasterSlave> reconcile(MySqlMasterSlave resource, Context context) {

        String namespace = Optional.of(resource.getMetadata().getNamespace()).orElse("default");
        String name = resource.getMetadata().getName();

        String mysqlRootPassword = resource.getSpec().getMysqlRootPassword();
        String mysqlReplicationUser = resource.getSpec().getMysqlReplicationUser();
        String mysqlReplicationPassword = resource.getSpec().getMysqlReplicationPassword();

        // 创建Master MYSQL
        Deployment deploymentMaster = loadYaml(Deployment.class, "deployment-master.yaml");

        withDeployMetadata(deploymentMaster, name, namespace, MASTER);

        withTemplateTable(deploymentMaster, name, MASTER);

        List<EnvVar> envVars = new ArrayList<>(3);
        envVars.add(new EnvVarBuilder()
                .withName("MYSQL_ROOT_PASSWORD")
                .withValue(mysqlRootPassword)
                .build());
        envVars.add(new EnvVarBuilder()
                .withName("MYSQL_USER")
                .withValue(mysqlReplicationUser)
                .build());
        envVars.add(new EnvVarBuilder()
                .withName("MYSQL_PASSWORD")
                .withValue(mysqlReplicationPassword)
                .build());
        withDeployEnv(deploymentMaster, envVars);

        withDeployVolumeConfigMap(deploymentMaster, configMapName(resource, MASTER));

        Service service = loadService("service-master.yaml", deploymentMaster, resource, namespace, MASTER);

        ConfigMap masterConfigMap = buildConfigMap(resource.getSpec().getMasterConfig(),
                configMapName(resource, MASTER), namespace);

        ConfigMap existingConfigMap = existingConfigMap(masterConfigMap);

        log.info("Creating or updating ConfigMap {} in {}", masterConfigMap.getMetadata().getName(), namespace);
        kubernetesClient.configMaps().inNamespace(namespace).createOrReplace(masterConfigMap);

        if (kubernetesClient.services().inNamespace(namespace).withName(service.getMetadata().getName()).get() == null) {
            log.info("Creating Service {} in {}", service.getMetadata().getName(), namespace);
            kubernetesClient.services().inNamespace(namespace).createOrReplace(service);
        }

        if (kubernetesClient.apps().deployments().inNamespace(namespace).withName(deploymentMaster.getMetadata().getName()).get() == null) {
            log.info("Creating Deployment {} in {}", deploymentMaster.getMetadata().getName(), namespace);
            kubernetesClient.apps().deployments().inNamespace(namespace).create(deploymentMaster);
        }

        if (existingConfigMap != null) {
            if (!StringUtils.equals(
                    existingConfigMap.getData().get("my.cnf"),
                    masterConfigMap.getData().get("my.cnf"))) {
                log.info("Restarting pods because MYSQL Config has changed in {}", namespace);
                restartPods(namespace, format("%s-%s", name, MASTER));
            }
        }

        // 创建Slave MYSQL
        Deployment deploymentSlave = loadYaml(Deployment.class, "deployment-slave.yaml");
        withDeployMetadata(deploymentSlave, name, namespace, SLAVE);
        withTemplateTable(deploymentSlave, name, SLAVE);

        ArrayList<EnvVar> envVarArrayList = new ArrayList<>(1);
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_ROOT_PASSWORD")
                .withValue(mysqlRootPassword)
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_MASTER_PASSWORD")
                .withValue(mysqlRootPassword)
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_SLAVE_PASSWORD")
                .withValue(mysqlRootPassword)
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_REPLICATION_USER")
                .withValue(mysqlReplicationUser)
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_REPLICATION_PASSWORD")
                .withValue(mysqlRootPassword)
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_MASTER_HOST")
                .withValue(format("mysql-master-svc.%s", namespace))
                .build());
        envVarArrayList.add(new EnvVarBuilder()
                .withName("MYSQL_SLAVE_HOST")
                .withValue(format("mysql-slave-svc.%s", namespace))
                .build());
        withDeployEnv(deploymentSlave, envVarArrayList);

        withDeployVolumeConfigMap(deploymentSlave, configMapName(resource, SLAVE));

        ConfigMap configMap = loadYaml(ConfigMap.class, "mysql-connector-script.yaml");
        configMap.getMetadata().setNamespace(namespace);

        ConfigMap slaveConfigMap =
                buildConfigMap(resource.getSpec().getSlaveConfig(), configMapName(resource,SLAVE), namespace);

        log.info("Creating or updating ConfigMap {} in {}", slaveConfigMap.getMetadata().getName(), namespace);
        kubernetesClient.configMaps().inNamespace(namespace).createOrReplace(slaveConfigMap);
        kubernetesClient.configMaps().inNamespace(namespace).createOrReplace(configMap);

        Service serviceSlave =
                loadService("service-slave.yaml", deploymentSlave, resource, namespace, SLAVE);

        if (kubernetesClient.services().inNamespace(namespace).withName(serviceSlave.getMetadata().getName()).get() == null) {
            log.info("Creating Service {} in {}", serviceSlave.getMetadata().getName(), namespace);
            kubernetesClient.services().inNamespace(namespace).createOrReplace(serviceSlave);
        }

        if (kubernetesClient.apps().deployments().inNamespace(namespace).withName(deploymentSlave.getMetadata().getName()).get() == null) {
            log.info("Creating Deployment {} in {}", deploymentSlave.getMetadata().getName(), namespace);
            kubernetesClient.apps().deployments().inNamespace(namespace).create(deploymentSlave);

            ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
            MySQLCommand mysqlConnectorCommand = new MySQLCommand(namespace, deploymentSlave, kubernetesClient, executorService, MYSQL_CONNECTOR_COMMAND);
            executorService.scheduleAtFixedRate(mysqlConnectorCommand, 30, 5, TimeUnit.SECONDS);
        }

        ConfigMap existingConfigMapSlave = existingConfigMap(slaveConfigMap);
        if (existingConfigMapSlave != null) {
            if (!StringUtils.equals(
                    existingConfigMapSlave.getData().get("my.cnf"),
                    slaveConfigMap.getData().get("my.cnf"))) {
                log.info("Restarting pods because MYSQL Config has changed in {}", namespace);
                restartPods(namespace, format("%s-%s", name, "slave"));
            }
        }

        return UpdateControl.updateStatus(resource);
    }


    private void withDeployMetadata(Deployment deployment, String name, String namespace, String mv) {
        deployment.getMetadata().setName(format("%s-%s", name, mv));
        deployment.getMetadata().setNamespace(namespace);
        deployment.getSpec().getSelector().getMatchLabels().put("app", format("%s-%s", name, mv));
    }

    private void withDeployVolumeConfigMap(Deployment deployment, String configMapName) {
        deployment
            .getSpec()
            .getTemplate()
            .getSpec()
            .getVolumes()
            .get(0)
            .setConfigMap(new ConfigMapVolumeSourceBuilder().withName(configMapName).build());
    }

    private void withTemplateTable(Deployment deployment, String name, String mv) {
        deployment
            .getSpec()
            .getTemplate()
            .getMetadata()
            .getLabels()
            .put("app", format("%s-%s", name, mv));
    }

    private void withDeployEnv(Deployment deployment, List<EnvVar> envVars) {
        deployment
            .getSpec()
            .getTemplate()
            .getSpec()
            .getContainers()
            .get(0)
            .getEnv().addAll(envVars);
    }

    private Service loadService(String serviceYaml, Deployment deployment, MySqlMasterSlave mysql,
                                String namespace, String mv) {
        Service service = loadYaml(Service.class, serviceYaml);
        service.getMetadata().setName(serviceName(mysql, mv));
        service.getMetadata().setNamespace(namespace);
        service.getSpec().setSelector(deployment.getSpec().getTemplate().getMetadata().getLabels());
        return service;
    }

    private ConfigMap buildConfigMap(String config, String configName, String namespace) {
        Map<String, String> data = new HashMap<>();
        data.put("my.cnf", config);
        return new ConfigMapBuilder()
                    .withMetadata(
                        new ObjectMetaBuilder()
                                .withName(configName)
                                .withNamespace(namespace)
                                .build())
                    .withData(data)
                    .build();
    }

    private ConfigMap existingConfigMap(ConfigMap configMap) {
        return kubernetesClient
                .configMaps()
                .inNamespace(configMap.getMetadata().getNamespace())
                .withName(configMap.getMetadata().getName())
                .get();
    }

    private void restartPods(String namespace, String label) {
        kubernetesClient
                .pods()
                .inNamespace(namespace)
                .withLabel("app", label)
                .delete();
    }

    private String serviceName(MySqlMasterSlave resource, String mv) {
        return format("%s-%s-svc", resource.getMetadata().getName(), mv);
    }

    private String deploymentName(MySqlMasterSlave resource, String mv) {
        return format("%s-%s", resource.getMetadata().getName(), mv);
    }

    private String configMapName(MySqlMasterSlave resource, String mv) {
        return format("%s-cnf-%s", resource.getMetadata().getName(), mv);
    }

    private void deleteConfigMap(MySqlMasterSlave resource, String configName) {
        log.info("Deleting ConfigMap {}", configName);
        Resource<ConfigMap> configMap =
                kubernetesClient
                        .configMaps()
                        .inNamespace(resource.getMetadata().getNamespace())
                        .withName(configName);
        if (configMap.get() != null) {
            configMap.delete();
        }
    }

    private void deleteDeployment(MySqlMasterSlave resource, String mv) {
        log.info("Deleting Deployment {}", deploymentName(resource, mv));
        RollableScalableResource<Deployment> deployment =
                kubernetesClient
                        .apps()
                        .deployments()
                        .inNamespace(resource.getMetadata().getNamespace())
                        .withName(deploymentName(resource, mv));
        if (deployment.get() != null) {
            deployment.cascading(true).delete();
        }
    }

    private void deleteService(MySqlMasterSlave resource, String mv) {
        log.info("Deleting Service {}", serviceName(resource, mv));
        ServiceResource<Service> service =
                kubernetesClient
                        .services()
                        .inNamespace(resource.getMetadata().getNamespace())
                        .withName(serviceName(resource, mv));
        if (service.get() != null) {
            service.delete();
        }
    }

    @Override
    public DeleteControl cleanup(MySqlMasterSlave resource, Context context) {

        deleteConfigMap(resource, configMapName(resource, MASTER));
        deleteConfigMap(resource, configMapName(resource, SLAVE));
        deleteConfigMap(resource, "mysql-connector-script");

        deleteDeployment(resource, MASTER);
        deleteDeployment(resource, SLAVE);

        deleteService(resource, MASTER);
        deleteService(resource, SLAVE);

        return DeleteControl.defaultDelete();
    }

    private <T> T loadYaml(Class<T> clazz, String yaml) {
        try (InputStream is = getClass().getResourceAsStream(yaml)) {
            return Serialization.unmarshal(is, clazz);
        } catch (IOException ex) {
            throw new IllegalStateException("Cannot find yaml on classpath: " + yaml);
        }
    }

    private static class MySQLCommand implements Runnable {

        private final Logger logger = LoggerFactory.getLogger(MySQLCommand.class);

        private final String namespace;
        private final Deployment deployment;
        private final KubernetesClient kubernetesClient;
        private final ScheduledExecutorService executorService;
        private final String command;

        public MySQLCommand(String namespace,
                            Deployment deployment,
                            KubernetesClient kubernetesClient,
                            ScheduledExecutorService executorService,
                            String command) {
            this.namespace = namespace;
            this.deployment = deployment;
            this.kubernetesClient = kubernetesClient;
            this.executorService = executorService;
            this.command = command;
        }

        @Override
        public void run() {
            int readyReplicas = kubernetesClient
                    .apps()
                    .deployments()
                    .inNamespace(namespace)
                    .withName(deployment.getMetadata().getName())
                    .get()
                    .getStatus()
                    .getReadyReplicas();

            logger.info("The resource {} in namespace [{}] ready replicas is: {}", deployment.getMetadata().getName(), namespace, readyReplicas);

            final Map<String, String> matchLabels = deployment.getSpec().getSelector().getMatchLabels();
            final List<Pod> pods = kubernetesClient.pods().inNamespace(namespace).withLabels(matchLabels).list().getItems();

            pods.stream().map(Pod::getMetadata).map(ObjectMeta::getName).forEach(this::podExec);

            if (readyReplicas == deployment.getSpec().getReplicas()) {
                executorService.shutdown();
            }
        }

        private void podExec(String podName) {
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            final ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(new byte[128]);
            final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            kubernetesClient
                    .pods()
                    .inNamespace(namespace)
                    .withName(podName)
                    .readingInput(byteArrayInputStream)
                    .writingOutput(byteArrayOutputStream)
                    .writingError(byteArrayOutputStream)
                    .usingListener(new SimpleListener(countDownLatch))
                    .exec("sh", "-c", command);
            try {
                countDownLatch.await();
                logger.info(byteArrayOutputStream.toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private static class SimpleListener implements ExecListener {

        private final CountDownLatch countDownLatch;

        private SimpleListener(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }
        @Override
        public void onOpen() {}

        @Override
        public void onFailure(Throwable t, ExecListener.Response failureResponse) {
            countDownLatch.countDown();
        }

        @Override
        public void onClose(int code, String reason) {
            countDownLatch.countDown();
        }
    }

}
